package hbase.user;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class TableTest {


    public static void main(String[] args){
        try {
            Configuration config = HBaseConfiguration.create();
            //Connection connection = ConnectionFactory.createConnection(config);
            //通过代码配置，无序设置配置文件，需要在代码中配置zk

            //Configuration configuration = new Configuration();
            config.set("hbase.rootdir", "hdfs://127.0.0.1:9000/hbase");
            config.set("hbase.zookeeper.quorum", "127.0.0.1");
            Connection connection = ConnectionFactory.createConnection(config);

            Admin admin = connection.getAdmin();
            String myNamespace = "laoshuming";

            NamespaceDescriptor[] listNamespaceDescriptors = admin.listNamespaceDescriptors();
            System.out.println(Arrays.toString(listNamespaceDescriptors));
            boolean exists=false;
            NamespaceDescriptor myNamespaceDes = null;
            for (NamespaceDescriptor namespaceDescriptor : listNamespaceDescriptors) {
                if(namespaceDescriptor.getName().equals(myNamespace)) {
                    myNamespaceDes = namespaceDescriptor;
                    exists=true;
                    break;
                }
            }
            if(!exists) {
                System.out.println("创建命名空间");
                myNamespaceDes = NamespaceDescriptor.create(myNamespace).build();
                admin.createNamespace(myNamespaceDes);
            }

            TableName tableName = TableName.valueOf("laoshuming:user");
            HTableDescriptor hTableDescriptor = null;
            if (!admin.tableExists(tableName)) {
                System.out.println("创建表");
                //2.2 如果表不存在就重新创建一个表
                //创建表描述对象
                hTableDescriptor = new HTableDescriptor(tableName);
                //通过表描述对象，添加列簇
                HColumnDescriptor infoHCloumn = new HColumnDescriptor("info");

                hTableDescriptor.addFamily(infoHCloumn);
                hTableDescriptor.addFamily(new HColumnDescriptor("score"));
                //通过admin创建表，需要传入表描述对象
                admin.createTable(hTableDescriptor);
            }else{
                System.out.println(tableName.getName()+"表已存在");
            }

            HTable userTable = new HTable(config, tableName.getName());

            Put tom = new Put(Bytes.toBytes("Tom"));
            tom.addColumn(Bytes.toBytes("info"),Bytes.toBytes("student_id"),Bytes.toBytes("20210000000001"));
            tom.addColumn(Bytes.toBytes("info"),Bytes.toBytes("class"),Bytes.toBytes("1"));
            tom.addColumn(Bytes.toBytes("score"),Bytes.toBytes("understanding"),Bytes.toBytes("75"));
            tom.addColumn(Bytes.toBytes("score"),Bytes.toBytes("programming"),Bytes.toBytes("82"));
            userTable.put(tom);

            Put jerry = new Put(Bytes.toBytes("Jerry"));
            jerry.addColumn(Bytes.toBytes("info"),Bytes.toBytes("student_id"),Bytes.toBytes("20210000000002"));
            jerry.addColumn(Bytes.toBytes("info"),Bytes.toBytes("class"),Bytes.toBytes("1"));
            jerry.addColumn(Bytes.toBytes("score"),Bytes.toBytes("understanding"),Bytes.toBytes("85"));
            jerry.addColumn(Bytes.toBytes("score"),Bytes.toBytes("programming"),Bytes.toBytes("67"));
            userTable.put(jerry);

            Put jack = new Put(Bytes.toBytes("Jack"));
            jack.addColumn(Bytes.toBytes("info"),Bytes.toBytes("student_id"),Bytes.toBytes("20210000000003"));
            jack.addColumn(Bytes.toBytes("info"),Bytes.toBytes("class"),Bytes.toBytes("2"));
            jack.addColumn(Bytes.toBytes("score"),Bytes.toBytes("understanding"),Bytes.toBytes("80"));
            jack.addColumn(Bytes.toBytes("score"),Bytes.toBytes("programming"),Bytes.toBytes("80"));
            userTable.put(jack);

            Put rose = new Put(Bytes.toBytes("Rose"));
            rose.addColumn(Bytes.toBytes("info"),Bytes.toBytes("student_id"),Bytes.toBytes("20210000000004"));
            rose.addColumn(Bytes.toBytes("info"),Bytes.toBytes("class"),Bytes.toBytes("2"));
            rose.addColumn(Bytes.toBytes("score"),Bytes.toBytes("understanding"),Bytes.toBytes("60"));
            rose.addColumn(Bytes.toBytes("score"),Bytes.toBytes("programming"),Bytes.toBytes("61"));
            userTable.put(rose);

            Put laoshuming = new Put(Bytes.toBytes("Laoshuming"));
            laoshuming.addColumn(Bytes.toBytes("info"),Bytes.toBytes("student_id"),Bytes.toBytes("G20210735020009"));
            laoshuming.addColumn(Bytes.toBytes("info"),Bytes.toBytes("class"),Bytes.toBytes("3"));
            laoshuming.addColumn(Bytes.toBytes("score"),Bytes.toBytes("understanding"),Bytes.toBytes("85"));
            laoshuming.addColumn(Bytes.toBytes("score"),Bytes.toBytes("programming"),Bytes.toBytes("85"));
            userTable.put(laoshuming);

            Scan scan = new Scan();
            ResultScanner results = userTable.getScanner(scan);

            System.out.println("扫描表");
            for(Result result : results){
                //解析result
                System.out.print(new String(result.getRow())+"\t");
                for(Cell cell : result.rawCells()){
                    System.out.print(Bytes.toString(CellUtil.cloneFamily(cell))+":"+Bytes.toString(CellUtil.cloneQualifier(cell))+":"+Bytes.toString(CellUtil.cloneValue(cell))+"\t");
                }
                System.out.println("");
            }

            Filter fiterNmae = new RowFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("Laoshuming")));
            scan.setFilter(fiterNmae);
            ResultScanner filterResults = userTable.getScanner(scan);
            System.out.println("搜索rowkey");
            for(Result result : filterResults){
                //解析result
                System.out.print(new String(result.getRow())+"\t");
                for(Cell cell : result.rawCells()){
                    System.out.print(Bytes.toString(CellUtil.cloneFamily(cell))+":"+Bytes.toString(CellUtil.cloneQualifier(cell))+":"+Bytes.toString(CellUtil.cloneValue(cell))+"\t");
                }
                System.out.println("");
            }

            List<Filter> filters = new ArrayList<Filter>();
            Filter colNamefilter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("understanding")));
            Filter valuefilter = new SingleColumnValueFilter(Bytes.toBytes("score"),Bytes.toBytes("understanding"),
                    CompareFilter.CompareOp.GREATER_OR_EQUAL,Bytes.toBytes("80"));
            filters.add(colNamefilter);
            filters.add(valuefilter);
            FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL,filters);
            scan.setFilter(filterList);
            ResultScanner colResults = userTable.getScanner(scan);
            System.out.println("搜索understanding分数大于80");
            for(Result result : colResults){
                //解析result
                System.out.print(new String(result.getRow())+"\t");
                for(Cell cell : result.rawCells()){
                    System.out.print(Bytes.toString(CellUtil.cloneFamily(cell))+":"+Bytes.toString(CellUtil.cloneQualifier(cell))+":"+Bytes.toString(CellUtil.cloneValue(cell))+"\t");
                }
                System.out.println("");
            }

            admin.disableTable(tableName);
            // 删除表
            admin.deleteTable(tableName);
            // 删除命名空间
            admin.deleteNamespace(myNamespace);
            admin.close();
            connection.close();
        }catch(Exception e){
            e.printStackTrace();
        }
    }
}
